package com.niudong.esdemo.pulsar;

import org.apache.pulsar.client.api.*;

public class Demo02 {

    public static void main(String[] args) throws PulsarClientException {
        PulsarClient client = PulsarClient.builder()
                .serviceUrl("pulsar://192.168.1.237:6650")
                .build();

        Consumer consumer = client.newConsumer()
                .topic("test_topic")
                .subscriptionName("my-subscription")
                .subscribe();

        // 然后你就可以发送消息到指定的broker 和topic上：

        while (true) {
            // Wait for a message
            Message msg = consumer.receive();

            try {
                // Do something with the message
                System.out.println("Message received: "+ new String(msg.getData()));

                // Acknowledge the message so that it can be deleted by the message broker
                consumer.acknowledge(msg);
            } catch (Exception e) {
                // Message failed to process, redeliver later
                consumer.negativeAcknowledge(msg);
            }
        }


    }
}
